package spring.cloud.tasks.executor.executor;

import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.executor.exception.TimeDiffIntolerableException;
import spring.cloud.tasks.executor.utils.HostUtil;
import spring.cloud.tasks.executor.utils.SystemEnvProperties;
import spring.cloud.tasks.executor.utils.log.LogEvents;
import spring.cloud.tasks.executor.utils.log.LogUtils;
import spring.cloud.tasks.executor.utils.zookeeper.Zookeeper;

@Slf4j
@Setter
@Getter
/**
 * Executor and Task
 */
public class ExecutorAndTaskInitService {
    public static final int WAIT_JOBCLASS_ADDED_COUNT = 25;
    private static final int WAIT_FOR_IP_NODE_DISAPPEAR_COUNT = 150;
    private String executorId;
    private Zookeeper zookeeper;
    private String executorIpNodePath;
    //

    private ExecutorRestartService executorRestartService;
    private TaskInitService taskInitService;

    public ExecutorAndTaskInitService(@NonNull String executorId, Zookeeper zookeeper) {
        this.executorId = executorId;
        this.zookeeper = zookeeper;
        zookeeper.setExecutorId(executorId);
    }
    //##<!-我是优美的功能模块分割线->##

    /**
     * 注册Executor
     */
    public void registerExecutor() throws Exception {
        checkExecutorExists();
        setExecutorInfo();
    }

    /**
     * 启动前先检查本机与注册中心的时间误差秒数是否在允许范围和Executor是否已启用
     */
    private void checkExecutorExists() throws Exception {
        // 启动时检查本机与注册中心的时间误差秒数是否在允许范围
        String executorNodePath = ExecutorsNode.EXECUTORS_ROOT + "/" + executorId;
        try {
            long timeDiff = Math.abs(System.currentTimeMillis() - zookeeper.getCtime(executorNodePath + "/systemTime/current"));
            int maxTimeDiffSeconds = 60;
            if (timeDiff > maxTimeDiffSeconds * 1000L) {
                Long timeDiffSeconds = Long.valueOf(timeDiff / 1000);
                throw new TimeDiffIntolerableException(timeDiffSeconds.intValue(), maxTimeDiffSeconds);
            }
        } finally {
            String executorSystemTimePath = executorNodePath + "/systemTime/current";
            if (zookeeper.checkExists(executorSystemTimePath)) {
                zookeeper.delete(executorSystemTimePath);
            }
        }
        // 启动时检查Executor是否已启用（ExecutorName为判断的唯一标识）
        if (!zookeeper.checkExists(executorNodePath)) {
            zookeeper.persist(executorNodePath, "");
        } else {
            int count = 0;
            do {
                if (!zookeeper.checkExists(executorNodePath + "/ip")) {
                    return;
                }
                LogUtils.warn(log, LogEvents.ExecutorEvent.INIT, "{}/ip node found. Try to sleep and wait for this node disappear.", executorNodePath);
                Thread.sleep(100L);
            } while (++count <= WAIT_FOR_IP_NODE_DISAPPEAR_COUNT);
            throw new Exception("The executor (" + executorId + ") is running, cannot running the instance twice.");
        }
    }

    private void setExecutorInfo() throws Exception {
        String executorNodePath = ExecutorsNode.EXECUTORS_ROOT + "/" + executorId;
        executorIpNodePath = executorNodePath + "/ip";
        // 持久化ip
        zookeeper.persistEphemeralDataNode(executorIpNodePath, HostUtil.cachedIpAddress);
        // 持久化最近启动时间
        String lastBeginTimeNodePath = executorNodePath + "/lastBeginTime";
        zookeeper.persist(lastBeginTimeNodePath, String.valueOf(System.currentTimeMillis()));
        // 持久化clean Executor离线时，其zk节点信息是否被清理
        String executorCleanNodePath = executorNodePath + "/clean";
        zookeeper.persist(executorCleanNodePath, String.valueOf(SystemEnvProperties.VIP_SATURN_EXECUTOR_CLEAN));

        // add watcher for restart
        if (executorRestartService != null) {
            executorRestartService.stop();
        }
        executorRestartService = new ExecutorRestartService(executorId, zookeeper);
        executorRestartService.start();
    }


    /**
     * 注销Executor
     */
    public void unRegisterExecutor() {
        stopRestartExecutorService();
        removeExecutorIpNode();
    }

    private void stopRestartExecutorService() {
        try {
            if (executorRestartService != null) {
                executorRestartService.stop();
            }
        } catch (Throwable t) {
            LogUtils.error(log, LogEvents.ExecutorEvent.INIT, t.toString(), t);
        }
    }

    private void removeExecutorIpNode() {
        try {
            if (zookeeper != null && executorIpNodePath != null && zookeeper.isConnected()) {
                LogUtils.info(log, LogEvents.ExecutorEvent.SHUTDOWN, "{} is going to delete its ip node {}", executorId, executorIpNodePath);
                zookeeper.delete(executorIpNodePath);
            }
        } catch (Throwable t) {
            LogUtils.error(log, LogEvents.ExecutorEvent.INIT, t.toString(), t);
        }
    }
    //##<!-我是优美的功能模块分割线->##

    /**
     * 注册$/Jobs的watcher，响应添加作业的事件，初始化作业。注意，会响应已经存在的作业。
     */
    public void registerTaskWatcher() throws Exception {
        if (taskInitService != null) {
            taskInitService.shutdown();
        }
        taskInitService = new TaskInitService(this);
        taskInitService.start();
    }

    public void removeTaskId(String taskId) {
        if (taskInitService != null) {
            taskInitService.removeTaskId(taskId);
        }
    }

    /**
     * 销毁监听添加作业的watcher。关闭正在初始化作业的线程，直到其结束。
     */
    public void unregisterJobsWatcher() {
        if (taskInitService != null) {
            taskInitService.shutdown();
        }
    }


}
